package com.tensai.study.java.newconcept.typereactiveprogram.reactor.flux;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author Tensai
 */
@Slf4j
public class FluxTest {

    public static void main(String[] args) {
        range();
        err();
    }

    private static void range(){
        Flux.range(1, 100)
                .timeout(Flux.never(), v -> Flux.never())
                .subscribe(item->{
                    try {
                        Thread.sleep(200);
                        System.out.println(item);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
    }

    private static void err(){
        Flux.range(1, 10)
                // 过滤
                .map(item -> {
                    // 偶数翻倍
                    if(item == 7) { // 到7进行异常抛出
                        throw new RuntimeException();
                    }
                    return item;
                })
                // 消费
                .subscribe(item -> System.out.print(item + " "), info -> System.out.print("异常"),
                        () -> System.out.println("完成"));
// 1 2 3 4 5 6 异常

    }

    /**
     * 通过元素创建流 元素的改变不影响流
     */
    @Test
    public void test(){
        List<Integer> list = IntStream.range(1, 20).boxed().collect(Collectors.toList());

        Flux.fromIterable(list).subscribe(it->{
            System.out.println(it);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        list.clear();

        for (; ; ) {
        }
    }

    @Test
    public void immediateCancel() {
        AtomicReference<String> value = new AtomicReference<>();
        AtomicReference<Throwable> error = new AtomicReference<>();
        Disposable s = Flux.just("foo", "bar")
                .delayUntil(v -> Mono.just(1).map(it -> {
                    log.error(it.toString());
                    return it;
                }))
                .subscribe(log::info);
        //would be a NPE if trigger array wasn't pre-initialized


    }
}
